package com.atguigu.dga230315.governance.assess.calc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga230315.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga230315.governance.assess.Assessor;
import com.atguigu.dga230315.governance.bean.AssessParam;
import com.atguigu.dga230315.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga230315.governance.utils.SqlParse;
import com.atguigu.dga230315.meta.bean.TableMetaInfo;
import com.atguigu.dga230315.meta.constant.MetaConst;
import com.google.common.collect.Sets;

import lombok.Getter;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.ParseException;
import java.util.*;

@Component("SIMPLE_PROCESS")
public class SimpleProcessAssess extends Assessor {
    @Override
    protected void assess(GovernanceAssessDetail assessDetail, AssessParam assessParam) throws ParseException {
        //sql语句没有任何join\groupby\ 非分区字段的where过滤  ，以上情况给0分，其余给10分	任务信息
        //思路：首先提取sql语句，然后根据sql语句进行分析，先分析有没有join group union...复杂操作 如果有这些操作的话那么就不是简单处理，如果没有这些操作 再去sql中获取过滤字段 然后 再获取查询的表，根据过滤在段和要查询表的分区字段对比，进而得到过滤字段是否都是分区字段 如果是的话那么就是简单处理

        //对ods层的表不做考评 并将考评日期没运行过的表不做考评
        if (assessParam.getTableMetaInfo().getTableMetaInfoExtra().getDwLevel().equals(MetaConst.DW_LEVEL_ODS)||assessParam.getTDsTaskDefinition()==null){
            return;
        }
        //1.提取hql
        TDsTaskDefinition tDsTaskDefinition = assessParam.getTDsTaskDefinition();
        String hql = tDsTaskDefinition.getSql();

        //2.解析hql
        MydisPatcher dispatcher = new MydisPatcher();
        SqlParse.sqlParse(dispatcher,hql);

        assessDetail.setAssessComment("涉及操作："+dispatcher.processSet+"过滤字段："+dispatcher.whereFieldSet+"查询的表："+dispatcher.fromTables);

        //获取当前考评表的库名
        String schemaName = assessParam.getTableMetaInfo().getSchemaName();

        //3.根据解析到的结果进行判断看是否是简单处理
        if (dispatcher.processSet.size()>0){
            //不是简单处理
            return;
        }else {
            //没有涉及到复杂操作 那么就需要判断 过滤字段是否都是分区字段
            //4.获取过滤字段
            Set<String> whereFieldSet = dispatcher.whereFieldSet;

            //5.获取查询的表
            Set<String> fromTablesName = dispatcher.fromTables;

            //6.根据查询的表的表名获取到分区字段
            for (String tableName : fromTablesName) {
                String fullTableName;
                if (tableName.contains(".")){
                    fullTableName = tableName;
                }else {
                    fullTableName = schemaName + "." + tableName;
                }
                //根据全表名获取到查询的表对应的TableMetaInfo
                TableMetaInfo fromTableMetaInfo = assessParam.getFullNameToTableMetaInfoMap().get(fullTableName);

                //获取到查询表的分区字段
                String partitionColNameJson = fromTableMetaInfo.getPartitionColNameJson();
                //将上面的字符串转为List<JsonObject>
                List<JSONObject> partitionColList = JSON.parseArray(partitionColNameJson, JSONObject.class);

                //定义计数器 用来计数过滤字段为分区字段的个数
                Integer sameColCount = 0;

                //双层for循环先遍历过滤字段 再遍历分区字段 看是否相同
                for (String whereField : whereFieldSet) {
                    for (JSONObject partitionObj : partitionColList) {
                        //提取分区字段名
                        String partitionField = partitionObj.getString("name");
                        if (whereField.equals(partitionField)){
                            sameColCount++;
                        }
                    }
                }

                //看计数器的个数是否等于过滤字段集合的元数个数 如果相等的话 那么证明所有过滤字段都是分区字段
                if (sameColCount==whereFieldSet.size()){
                    assessDetail.setAssessScore(BigDecimal.ZERO);
                    assessDetail.setAssessProblem("是简单处理");
                }
            }
        }

    }

    private static class MydisPatcher implements Dispatcher{

        //收集复杂操作的集合
        @Getter
        Set<String> processSet=new HashSet<>();

        //收集过滤字段
        @Getter
        Set<String> whereFieldSet=new HashSet<>();

        //收集查询的表
        @Getter
        Set<String> fromTables=new HashSet<>();

        //都有哪些复杂操作
        Set<Integer> realProcessSet= Sets.newHashSet(HiveParser.TOK_JOIN,  //join 包含通过where 连接的情况
                HiveParser.TOK_GROUPBY,       //  group by
                HiveParser.TOK_LEFTOUTERJOIN,       //  left join
                HiveParser.TOK_RIGHTOUTERJOIN,     //   right join
                HiveParser.TOK_FULLOUTERJOIN,     // full join
                HiveParser.TOK_FUNCTION,     //count(1)
                HiveParser.TOK_FUNCTIONDI,  //count(distinct xx)
                HiveParser.TOK_FUNCTIONSTAR, // count(*)
                HiveParser.TOK_SELECTDI,  // distinct
                HiveParser.TOK_UNIONALL   // union
        );

        Set<String> operators=Sets.newHashSet("=",">","<",">=","<=" ,"<>"  ,"like"); // in / not in 属于函数计算
        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {

            ASTNode astNode = (ASTNode) nd;

            //收集 goupby  join  union 这种复杂操作
            //获取节点的类型判断当前遍历到的这个节点类型是否属于realProcessSet这个集合中的元素
            if (realProcessSet.contains(astNode.getType())){
                processSet.add(astNode.getText());
            }

            //收集 过滤字段
            //先判断当前遍历到的节点文本是不是= > < >= 这种操作符 是的话那么这个节点下面的子节点才会有过滤字段
            if (operators.contains(astNode.getText())){
                //获取到当前遍历得到节点的 所有子节点
                ArrayList<Node> children = astNode.getChildren();
                //判断子节点是否有 点 如果有点的话证明 这个过滤字段带表名
                for (Node child : children) {
                    ASTNode astNodeChild = (ASTNode) child;
                    if (astNodeChild.getType()==HiveParser.DOT){
                        whereFieldSet.add(astNodeChild.getChild(1).getText());
                    } else if (astNodeChild.getType()==HiveParser.TOK_TABLE_OR_COL) {
                        whereFieldSet.add(astNodeChild.getChild(0).getText());
                    }
                }
            }

            //收集查询的表
            if (astNode.getType()==HiveParser.TOK_TABREF){
                //这个节点下面的节点的表是要查询的表
                ASTNode nodeChild = (ASTNode) astNode.getChild(0); //TOK_TABNAME
                ArrayList<Node> nodeChildChildren = nodeChild.getChildren();
                if (nodeChildChildren.size()>1){
                    //有库名
                    fromTables.add(nodeChild.getChild(0).getText()+"."+nodeChild.getChild(1).getText());
                }else {
                    fromTables.add(nodeChild.getChild(0).getText());
                }
            }

            return null;
        }
    }
}
